package datastream.api.operator.one.source.log;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import datastream.api.operator.one.pojo.log.InputModel;

import java.util.ArrayList;
import java.util.List;

public class LogSource implements SourceFunction<InputModel> {
    @Override
    public void run(SourceContext<InputModel> ctx) throws Exception {
        List one = new ArrayList<>();
        one.add(new Tuple2<String, String>("页面1", "按钮1"));
        one.add(new Tuple2<String, String>("页面1", "按钮2"));
        ctx.collect(InputModel.builder()
                .username("张三")
                .batchLog(one)
                .build());
        Thread.sleep(1000);

        List two = new ArrayList<>();
        two.add(new Tuple2<String, String>("页面1", "按钮1"));
        two.add(new Tuple2<String, String>("页面2", "按钮2"));
        two.add(new Tuple2<String, String>("页面1", "按钮3"));
        ctx.collect(InputModel.builder()
                .username("李四")
                .batchLog(two)
                .build());
    }

    @Override
    public void cancel() {

    }
}
